package heartbeat;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;

import sdm.repackaged.org.json.JSONObject;

public class Heartbeat implements Runnable
{
	protected List<TcpClient> mClients;
	protected List<TcpClient> mClientsToClose;
	protected LinkedBlockingQueue<Callback> mMessages;
	protected boolean mStopped = false;
	protected static final String VERSION = "1.0.0";
	protected static final int LIBRARY_VERSION = 1;
	protected static final int LIBRARY_CHANNEL_LIST = 2;
	protected static final int LIBRARY_AUTHENTICATION = 5;
	protected static final int PREDEFINED_MAX = 6;
	protected static Logger sLogger = Logger.getLogger(Heartbeat.class);
	
	public Heartbeat()
	{
		mClients = new ArrayList<TcpClient>();
		mClientsToClose = new ArrayList<TcpClient>();
		mMessages = new LinkedBlockingQueue<Callback>();
	}
	
	public String channelName(final TcpClient pClient, int channelNumber)
	{
		return pClient.channelName(channelNumber);
	}
	
	public Integer channelNumber(final TcpClient pClient, String channelName)
	{
		return pClient.channelNumber(channelName);
	}
	
	public void close(final TcpClient pClient)
	{
		Callback callback = new Callback()
		{
			@Override
			public void invoke()
			{
				doClose(pClient);
			}
		};
		
		mMessages.offer(callback);
	}
	
	public void sendAuthentication(final TcpClient pClient, final JSONObject authData)
	{
		Callback callback = new Callback()
		{
			@Override
			public void invoke()
			{
				doWrite(pClient, "Library_Authentication", authData);
			}
		};
		
		mMessages.offer(callback);
	}
	
	public void stop()
	{
		Callback callback = new Callback()
		{
			@Override
			public void invoke()
			{
				doStop();
			}
		};
		
		mMessages.offer(callback);
	}
	
	public void write(final TcpClient pClient, final Packet msg)
	{
		Callback callback = new Callback()
		{
			@Override
			public void invoke()
			{
				doWrite(pClient, msg);
			}
		};
		
		mMessages.offer(callback);
	}
	
	public void write(final TcpClient pClient, final String channelName, final JSONObject message)
	{
		//If a channels message hasn't come in yet, queue this.
		if(!pClient.channelMapValid())
		{
			Callback cb = new Callback()
			{
				public void invoke()
				{
					pClient.addChannelMessageToQueue(channelName, message);
				}
			};
			
			mMessages.offer(cb);
			return;
		}
		
		//Write the message.
		Callback cb = new Callback()
		{
			public void invoke()
			{
				doWrite(pClient, channelName, message);
			}
		};
		
		mMessages.offer(cb);
	}
	
	public TcpClient addClient(String address, int port, boolean ssl) throws Exception
	{
		TcpClient newClient = new TcpClient(address, port, ssl);
		mClients.add(newClient);
		return newClient;
	}

	@Override
	public void run() 
	{
		while(true)
		{
			try 
			{
				Thread.sleep(100);
			} 
			catch (InterruptedException e) 
			{
				Thread.currentThread().interrupt();
			}
			
			//Exit if the heartbeat is stopped.
			if(mStopped)
			{
				for(TcpClient pClient: mClients)
				{
					pClient.close();
				}
				mClients.clear();
				return;
			}
			
			//Remove the clients that have been closed.
			for(TcpClient pClient : mClientsToClose)
			{
				pClient.close();
				mClients.remove(pClient);
			}
			
			//Clear the close list.
			mClientsToClose.clear();
			
			//Empty the message queue.
			while(!mMessages.isEmpty())
			{
				Callback cb = mMessages.poll();
				if(cb == null) 
					break;
				
				cb.invoke();
			}
				
			
			//Try reading from each of the sockets to see if there's an incoming message.
			for(TcpClient pClient : mClients)
			{
				try
				{
					Packet p = pClient.read();
				
					//No message, continue.
					if(p == null)
						continue;
				
				
					messageReceived(pClient, p);
				}
				catch(IOException e)
				{
					sLogger.error(e);
					doClose(pClient);
				}
			}
		}
	}
	
	public void setOnAuthenticationCallback(final TcpClient pClient, final PCallback<MessageCBType> pCallback)
	{
		Callback cb = new Callback()
		{

			@Override
			public void invoke() 
			{
				pClient.setOnAuthenticationCallback(pCallback);	
			}
			
		};
		
		mMessages.offer(cb);
	}
	
	public void setOnClientDisconnectCallback(final TcpClient pClient, final PCallback<TcpClient> pCallback)
	{
		Callback cb = new Callback()
		{

			@Override
			public void invoke() 
			{
				pClient.setOnClientDisconnectCallback(pCallback);	
			}
			
		};
		
		mMessages.offer(cb);
	}
	
	public void setOnChannelsReceivedCallback(final TcpClient pClient, final PCallback<ChannelsCBType> pCallback)
	{
		Callback cb = new Callback()
		{

			@Override
			public void invoke() 
			{
				pClient.setOnChannelsReceivedCallback(pCallback);	
			}
			
		};
		
		mMessages.offer(cb);
	}
	
	public void setOnMessageCallback(final TcpClient pClient, final PCallback<MessageCBType> pCallback)
	{
		Callback cb = new Callback()
		{

			@Override
			public void invoke() 
			{
				pClient.setOnMessageCallback(pCallback);	
			}
			
		};
		
		mMessages.offer(cb);
	}
	
	public void subscribeTo(final TcpClient pClient, final String channelName) 
	{
		Callback cb = new Callback()
		{
			public void invoke()
			{
				doSubscribeTo(pClient, channelName);
			}
		};
		
		mMessages.offer(cb);
	}
	
	public void subscribeTo(final TcpClient pClient, final List<String> channelNames)
	{
		Callback cb = new Callback()
		{
			public void invoke()
			{
				doSubscribeTo(pClient, channelNames);
			}
		};
		
		mMessages.offer(cb);
	}
	
	private void doSubscribeTo(final TcpClient pClient, String channelName)
	{
		List<String> channelNameList = new ArrayList<String>();
		channelNameList.add(channelName);
		doSubscribeTo(pClient, channelNameList);
	}
	
	private void doSubscribeTo(final TcpClient pClient, final List<String> channelNames)
	{
		try
		{
			pClient.subscribeTo(channelNames);
		}
		catch(IOException e)
		{
			sLogger.error(e);
			doClose(pClient);
		}
	}
	
	protected void messageReceived(TcpClient pClient, Packet pPacket) throws IOException
	{
		int channel = pPacket.channel();
		if(channel == LIBRARY_VERSION)
		{
			if(!versionCheck(pPacket.message()))
			{
				System.out.println("Version check failed");
				doClose(pClient);
			}
			
			return;
		}
		if(channel == LIBRARY_CHANNEL_LIST)
		{
			handleChannelListReceived(pClient, pPacket.message());
			return;
		}
		
		if(channel == LIBRARY_AUTHENTICATION)
		{
			pClient.invokeOnAuthenticationCallback(new MessageCBType(pClient, pPacket));
		}
		if(channel <= PREDEFINED_MAX)
		{
			return;
		}
		
		pClient.invokeOnMessageCallback(new MessageCBType(pClient, pPacket));
	}
	
	private void handleChannelListReceived(TcpClient pClient, JSONObject pMessage) throws IOException
	{
		//Fill the client's channel map.
		pClient.fillChannelMap(pMessage);
		
		//Empty the queue of channel name/message pairs.
		pClient.emptyChannelMessageQueue();
		
		//Invoke the channels received callback.
		pClient.invokeOnChannelsReceivedCallback();
	}
	
	//Private do functions that only get invoked on the executor thread.
	protected void doClose(TcpClient pClient)
	{
		mClientsToClose.add(pClient);
	}
	
	private void doStop()
	{
		mStopped = true;
	}
	
	private void doWrite(TcpClient pClient, Packet msg)
	{
		try
		{
			pClient.write(msg);
		}
		catch(IOException e)
		{
			sLogger.error(e);
			doClose(pClient);
		}
	}
	
	private void doWrite(TcpClient pClient, String channelName, JSONObject pMessage)
	{
		try
		{
			pClient.write(channelName, pMessage);
		}
		catch(IOException e)
		{
			sLogger.error(e);
			doClose(pClient);
		}
	}
	
	private boolean versionCheck(JSONObject message)
	{
		try 
		{
			String serverVersion = message.getString("version");
			String[] serverxyz = serverVersion.split("\\.");
			String[] clientxyz = VERSION.split("\\.");
			
			if(serverxyz.length != 3 || clientxyz.length != 3 || !serverxyz[0].equalsIgnoreCase(clientxyz[0]) || !serverxyz[1].equalsIgnoreCase(clientxyz[1]))
				return false;
			
			return true;
		}
		catch(Exception e)
		{
			sLogger.error(e);
			return false;
		}
	}
}

